package com.jd.rapid.rpc.client;


import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 客户端链接管理器
 * @author  yls
 * @date 2021-07-29
 */
@Slf4j
public class RpcConnectManager {

    //volatile  多线程，这个对象这个内容可能会改变
    private static volatile  RpcConnectManager RPC_CONNECT_MANAGER =new RpcConnectManager();

    private long connectedTimoutMills=6000;

    private volatile  boolean isRunning=true;

    private volatile AtomicInteger handlerIdx=new AtomicInteger(0);

    private RpcConnectManager(){

    }

    public static  RpcConnectManager getInstance(){
        return RPC_CONNECT_MANAGER;
    }


    /****一个连接的地址，对应一个实际的业务处理器(client)****/
    private Map<InetSocketAddress,RpcClientHandler>  connetedHandlerMap=new ConcurrentHashMap<>();

    /*****所有连接成功的地址所对应的任务执行器列表*********************/
    private CopyOnWriteArrayList<RpcClientHandler> connetedHandlerList=new CopyOnWriteArrayList<>();


    /*****用于异步的提交连接请求的线程池************/
    private ThreadPoolExecutor threadPoolExecutor=new ThreadPoolExecutor(16,16,
            60, TimeUnit.SECONDS,new ArrayBlockingQueue<>(65536));

    //默认是cpu*2的大小
    private EventLoopGroup eventLoopGroup=new NioEventLoopGroup(4);

    private ReentrantLock connectedLock=new ReentrantLock();

    private Condition connectedCondition=connectedLock.newCondition();

    //1、异步创建连接、连接线程池,真正的发起连接，连接失败监听，连接成功
    //2、对于连接进来的资源做一个缓冲（做一个管理）updateConnectedServer


    /**
     * 发起连接
     * @param serverAddress
     */
    public  void connect(final String serverAddress){
        List<String> allServerAddress=Arrays.asList(serverAddress.split(","));
        updateConnectedServer(allServerAddress);
    }


    /**
     * 更新缓存信息并异步发起连接
     * 192.168.11.111:8765,192.168.11.111:8765
     * @param allServerAddress
     */
    public void updateConnectedServer( List<String> allServerAddress){
        if(CollectionUtils.isNotEmpty(allServerAddress)){
            //hashSet不重复  1、解析allServerAddress地址，并且临时存储到我们的newAllServerNodeSet Hashset集合中
            HashSet<InetSocketAddress>  newAllServerNodeSet=new HashSet<>();
            for(int i=0;i<allServerAddress.size();i++){
               String[] array=allServerAddress.get(i).split(":");
               if(array.length==2){
                   String host=array[0];
                   int port=Integer.parseInt(array[1]);
                   final InetSocketAddress remotePeer=new InetSocketAddress(host,port);
                   newAllServerNodeSet.add(remotePeer);
               }
            }

            //2、调用建立连接方法发起远程连接操作
            //发起连接？？怎么发起连接，异步的发起连接
            //有些连接已经存在？？不再重新发起连接
            for(InetSocketAddress serverNodeAddress:newAllServerNodeSet){
                  if(!connetedHandlerMap.keySet().contains(serverNodeAddress)) {
                      connectAsync(serverNodeAddress);
                  }
            }

            //3、allServerAddress列表里不存在的连接地址，那么我需要从缓冲中移除,移除不存在的列表
            //新的连接192.168.11.111:8765，92.168.11.111:8765，92.168.11.111:8765（2个）
            //老的连接1192.168.11.111:8765，92.168.11.111:8765，92.168.11.113:8765（3个）
            for(int i=0;i<connetedHandlerList.size();i++){
                RpcClientHandler rpcClientHandler=connetedHandlerList.get(i);
                SocketAddress socketAddress=rpcClientHandler.getRemotePeer();
                if(!newAllServerNodeSet.contains(socketAddress)){
                    log.info("remove invalid server node "+socketAddress);
                    RpcClientHandler handler= connetedHandlerMap.get(socketAddress);
                    if(handler!=null) {
                        handler.close();
                        connetedHandlerMap.remove(socketAddress);
                    }
                    connetedHandlerList.remove(rpcClientHandler);
                }
            }
        }else{
            //添加告警
            log.error("no available server address! ");
            //清除所以的缓冲信息
            clearConnected();
        }
    }

    /**
     * 异步发起连接的方法
     * @param serverNodeAddress
     */
    private void connectAsync(InetSocketAddress serverNodeAddress){
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                Bootstrap b=new Bootstrap();
                b.group(eventLoopGroup)
                 .channel(NioSocketChannel.class)
                 .option(ChannelOption.TCP_NODELAY,true)
                 //初始化
                 .handler(new RpcClientInitializer());
                //连接
                connect(b,serverNodeAddress);
            }
        });
    }


    /**
     * 发起连接
     * @param bootstrap
     * @param serverNodeAddress
     */
    public  void connect(final Bootstrap bootstrap,InetSocketAddress serverNodeAddress){
         //1、 真正的建立连接
        ChannelFuture channelFuture= bootstrap.connect(serverNodeAddress);
        //2、 连接失败的时候添加监听，清楚资源后进行发起重连操作
        channelFuture.channel().closeFuture().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                log.info("channelFuture.channel close operationComplete ,remote peer="+serverNodeAddress);
                channelFuture.channel().eventLoop().schedule(new Runnable() {
                    @Override
                    public void run() {
                        //每隔3秒钟重新发起连接
                        log.warn(" connec fial,to reconnect! per 3 scoends");
                        clearConnected();
                        connect(bootstrap,serverNodeAddress);
                    }
                },3, TimeUnit.SECONDS);
            }
        });
        //3、连接成功的时候添加监听，把我们的新连接放入缓冲中
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if(channelFuture.isSuccess()){
                        log.info("successfully connect to remote server="+serverNodeAddress);
                        RpcClientHandler handler=channelFuture.channel().pipeline().get(RpcClientHandler.class);
                        //添加到缓冲中
                        addHander(handler);
                    }else{
                        log.error("error: connect to remote server="+serverNodeAddress);
                    }
            }
        });

    }


    /**
     * 添加RpcClientHandler到指定的缓冲中
     * connetedHandlerMap & connetedHandlerList
     * @param handler
     */
    private void addHander(RpcClientHandler handler) {
        //list add  handler
        connetedHandlerList.add(handler);
//        InetSocketAddress inetSocketAddress= (InetSocketAddress) handler.getRemotePeer();
        InetSocketAddress inetSocketAddress= (InetSocketAddress) handler.getChannel().remoteAddress();
        //绑定map  inetSocketAddress-->RpcClientHandler
        connetedHandlerMap.put(inetSocketAddress,handler);
        //signalAvailableHandler唤醒可用的业务执行器
        //比如队列的take操作，阻塞，put的时候唤醒
        signalAvailableHandler();
    }



    /**
     *
     * 连接失败时，及时的释放资源，清空缓存
     * 先删除所有的connetedHandlerMap中的数据
     * 然后再清空connectedHandlerList中的数据
     */
    private void clearConnected() {
        for(final RpcClientHandler rpcClientHandler:connetedHandlerList){
            //通过rpcClientHandler找到具体的remote peer,进行指定的移除connetedHandlerMap
           SocketAddress socketAddress= rpcClientHandler.getRemotePeer();
           RpcClientHandler handler= connetedHandlerMap.get(socketAddress);
           if(handler!=null) {
               handler.close();
               connetedHandlerMap.remove(socketAddress);
           }
        }
        connetedHandlerList.clear();
    }



    /**
     * 唤醒另一端的线程（阻塞的状态中）告知有新连接接入
     */
    private void signalAvailableHandler() {
        connectedLock.lock();
        try{
            //唤醒
            connectedCondition.signalAll();
        }finally {
            connectedLock.unlock();
        }
    }

    private boolean waitingForAvailableHandler() throws InterruptedException {
        connectedLock.lock();
        try{
           return connectedCondition.await(this.connectedTimoutMills,TimeUnit.MILLISECONDS);
        }finally {
            connectedLock.unlock();
        }
    }


    /**
     * 阶段二
     * RpcClientHandler 选择一个实际的业务处理器
     * @return
     */
    public RpcClientHandler chooseHandler(){
        //取出列表，多线程之间线程安全，最好调用克隆一份
        //复制出来的没有线程安全
       CopyOnWriteArrayList<RpcClientHandler>  handlers= (CopyOnWriteArrayList<RpcClientHandler>) this.connetedHandlerList.clone();
       int size=handlers.size();
        while (isRunning &&size<=0){
            try {
                boolean avaiable=waitingForAvailableHandler();
                if(avaiable){
                    handlers= (CopyOnWriteArrayList<RpcClientHandler>) this.connetedHandlerList.clone();
                    size=handlers.size();
                }else{
                    log.info("no avaiable handlers,wait ......");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                log.error("wating for avialbe  node is Interrupted ");
                throw  new RuntimeException("no connet any server");
            }
        }
        if(!isRunning){
            return null;
        }
        //最终使用取模方式取得其中一个业务处理器进行实际的业务处理
        return handlers.get((handlerIdx.getAndAdd(1)+size)%size);
    }


    /**
     * 关闭
     */
    public void stop(){
        isRunning=false;
        for(int i=0;i<connetedHandlerList.size();i++){
            RpcClientHandler rpcClientHandler=connetedHandlerList.get(i);
            rpcClientHandler.close();
        }
        //在这里要调用一下唤醒操作,让chooseHandler的方法继续运行，不在阻塞因为isRunning=fasle
        //唤醒正在阻塞的线程
        signalAvailableHandler();
        threadPoolExecutor.shutdown();
        eventLoopGroup.shutdownGracefully();
    }


    /**
     * 发起重连方法，需要吧对应的资源释放
     * @param handler
     * @param socketAddress
     */
    public void reconnect(final RpcClientHandler handler ,final  SocketAddress socketAddress){
            if(handler!=null){
                handler.close();
                connetedHandlerList.remove(handler);
                connetedHandlerMap.remove(socketAddress);
            }
            connectAsync((InetSocketAddress) socketAddress);
    }
}
